package com.bage.finance.biz.mq;

import com.bage.common.exception.BizException;
import com.bage.finance.biz.constant.MqConstant;
import com.bage.finance.biz.domain.MqMsg;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.annotation.SelectorType;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Service;

/**
 * 消息消费方
 * 1.如果两个消费者group和topic都一样,则二者轮询接收消息
 * 2.如果两个消费者topic一样,而group不一样,则消息变成广播机制
 * RocketMQListener<>泛型必须和接收的消息类型相同
 */
@Slf4j
@Service
@RequiredArgsConstructor
@RocketMQMessageListener(
        topic = MqConstant.TOPIC_VOUCHER,   // topic:消息的发送者使用同一个topic
        consumerGroup = "consumer-group-1", // group:不用和生产者group相同(在RocketMQ中消费者和发送者组没有关系)
        selectorExpression = "*",   // tag:设置为"*"时,表示全部,多个以|分隔,如果tag1|tag2
        selectorType = SelectorType.TAG,    // 指定需要消费的TAG
        messageModel = MessageModel.CLUSTERING  // 消费模式:默认CLUSTERING(负载均衡)[BROADCASTING:广播机制]
)
public class SaveVoucherConsumer implements RocketMQListener<MqMsg> {

    final ApplicationContext applicationContext;

    final ObjectMapper objectMapper;

    @Override
    public void onMessage(MqMsg mqMsg) {
        try {
            Object object = objectMapper.readValue(mqMsg.getMsg(), Class.forName(mqMsg.getMsgClassName()));
            applicationContext.publishEvent(object);
        } catch (Exception e) {
            throw new BizException("消费异常", e);
        }
    }
}
